package algorithms;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;

/**
 * Each node will emit it's page rank to its neighbors along with its total
 * number out links.
 * 
 */
public class PageRankMapper extends Mapper<LongWritable, Text, Text, Text> {

	public Map<String, Double> inMapperCombiner;
	boolean useIMC = false;
	
	// public void map(LongWritable key, Text value, OutputCollector<Text, Text>
	// output, Reporter reporter)
	public void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		
		if (useIMC) {
			imcMap(key, value, context);
		} else{
		
			// index.html <pagerank> a.html b.html c.html ...
			String[] nodesAndPageRank = value.toString().split("\t");
	
			// Create the output value
			Integer outlinks = nodesAndPageRank.length - 2;
			Double pageRank = new Double(nodesAndPageRank[1]) / outlinks;
			//DoubleWritable pageRankOutput = new DoubleWritable(pageRank);
			
			/*Text outputValue = new Text(nodesAndPageRank[0] + "\t"
					+ nodesAndPageRank[1] + "\t" + outlinks.toString()); */
	
			String neighbors = "";
			for (int i = 2; i < nodesAndPageRank.length; i++) {
				// output.collect(new Text(nodesAndPageRank[i]), outputValue);
				context.write(new Text(nodesAndPageRank[i]), new Text(pageRank.toString()));
				neighbors += nodesAndPageRank[i] + "\t";
			}
			
			// Write out vertex state as well.
			context.write(new Text(nodesAndPageRank[0]), new Text("VERTEX_STATE\t" + neighbors.trim()));
		}
	}
	
	@Override
	protected void setup(Context context) {
		useIMC = context.getConfiguration().getBoolean("imc", false);
		inMapperCombiner = new HashMap<String, Double>();
	}
	

	public void imcMap(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		// index.html <pagerank> a.html b.html c.html ...
		String[] nodesAndPageRank = value.toString().split("\t");

		// Create the output value
		Integer outlinks = nodesAndPageRank.length - 2;
		Double pageRank = new Double(nodesAndPageRank[1]) / outlinks;

		String neighbors = "";
		for (int i = 2; i < nodesAndPageRank.length; i++) {
			// output.collect(new Text(nodesAndPageRank[i]), outputValue);
			//context.write(new Text(nodesAndPageRank[i]), new Text(pageRank.toString()));
			
			Double pr = inMapperCombiner.get(nodesAndPageRank[i]);
			if (pr == null) {
				pr = pageRank;
			} else {
				pr = pr + pageRank;
			}
			inMapperCombiner.put(nodesAndPageRank[i], pr);
			
			neighbors += nodesAndPageRank[i] + "\t";
		}
		
		// Write out vertex state as well.
		context.write(new Text(nodesAndPageRank[0]), new Text("VERTEX_STATE\t" + neighbors.trim()));
	}
	
	@Override
	protected void cleanup(Mapper.Context context) throws IOException, InterruptedException {
		for (String node : inMapperCombiner.keySet()) {
			context.write(new Text(node), new Text(inMapperCombiner.get(node).toString()));
		}
	}
}
